import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @author zhangli
 * @version V1.0
 * @create 2019-09-09 14:53
 **/
public class MsgConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "127.0.0.1:9092");
        props.put("group.id", "joe1");//组id 如果重新定义数据就会重新进行采集
        props.put("enable.auto.commit", "false");//如果为真，consumer所消费消息的offset将会自动的同步到zookeeper。如果消费者死掉时，由新的consumer使用继续接替
        props.put("auto.commit.interval.ms", "1000");//consumer向zookeeper提交offset的频率
        props.put("auto.offset.reset", "earliest");//如果存在已经提交的offest时,不管设置为earliest 或者latest 都会从已经提交的offest处开始消费
        props.put("session.timeout.ms", "120000");//超时时间
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅topic，可以为多个用,隔开，此处订阅了"test-partition-1", "2019gnyw"这两个主题 中的一个
        consumer.subscribe(Arrays.asList("2019gnyw"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }

}
